package org.trade.pay.service.impl;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.trade.common.constant.MQEnums;
import org.trade.common.constant.TradeEnums;
import org.trade.common.exception.AceMQException;
import org.trade.common.protocol.mq.PaidMQ;
import org.trade.common.protocol.pay.CallbackPaymentReq;
import org.trade.common.protocol.pay.CallbackPaymentRes;
import org.trade.common.protocol.pay.CreatePaymentReq;
import org.trade.common.protocol.pay.CreatePaymentRes;
import org.trade.common.rocketmq.AceMQProducer;
import org.trade.common.util.IDGenerator;
import org.trade.entity.TradeMqProducerTemp;
import org.trade.entity.TradePay;
import org.trade.entity.TradePayExample;
import org.trade.mapper.TradeMqProducerTempMapper;
import org.trade.mapper.TradePayMapper;
import org.trade.pay.service.IPayService;

import com.alibaba.fastjson.JSON;

public class PayServiceImpl implements IPayService {

	@Autowired
	private TradePayMapper tradePayMapper;

	/*
	 * 第三方支付平台回调我们之后，修改订单状态为已支付。节耦合
	 */
	@Autowired
	private AceMQProducer aceMQProducer;

	@Autowired
	private TradeMqProducerTempMapper tradeMqProducerTempMapper;

	private ExecutorService executorService = Executors.newFixedThreadPool(10);

	public CreatePaymentRes createPayment(CreatePaymentReq createPaymentReq) {
		CreatePaymentRes createPaymentRes = new CreatePaymentRes();
		createPaymentRes.setRetCode(TradeEnums.RetEnum.SUCCESS.getCode());
		createPaymentRes.setRetInfo(TradeEnums.RetEnum.SUCCESS.getDesc());
		try {
			TradePayExample payExample = new TradePayExample();
			payExample.createCriteria().andOrderIdEqualTo(createPaymentReq.getOrderId())
					.andIsPaidEqualTo(TradeEnums.YesNoEnum.YES.getCode());
			long count = this.tradePayMapper.countByExample(payExample);
			if (count > 0) {
				throw new Exception("该订单已支付");
			}
			String payId = IDGenerator.generatorUUID();
			TradePay tradePay = new TradePay();
			tradePay.setPayId(payId);
			tradePay.setOrderId(createPaymentReq.getOrderId());
			tradePay.setIsPaid(TradeEnums.YesNoEnum.NO.getCode());
			tradePay.setPayAmount(createPaymentReq.getPayAmount());
			tradePayMapper.insert(tradePay);
			System.out.println("创建支付订单成功:" + payId);
		} catch (Exception e) {
			createPaymentRes.setRetCode(TradeEnums.RetEnum.FAIL.getCode());
			createPaymentRes.setRetInfo(e.getMessage());
		}
		return createPaymentRes;
	}

	/*
	 * 回调支付接口，修改状态。发送可靠消息
	 */
	@Transactional
	public CallbackPaymentRes callbackPayment(CallbackPaymentReq callbackPaymentReq) {
		CallbackPaymentRes callbackPaymentRes = new CallbackPaymentRes();
		callbackPaymentRes.setRetCode(TradeEnums.RetEnum.SUCCESS.getCode());
		callbackPaymentRes.setRetInfo(TradeEnums.RetEnum.SUCCESS.getDesc());
		if (callbackPaymentReq.getIsPaid().equals(TradeEnums.YesNoEnum.YES.getCode())) {

			TradePay tradePay = tradePayMapper.selectByPrimaryKey(callbackPaymentReq.getIsPaid());
			if (tradePay == null) {
				throw new RuntimeException("未找到支付订单");
			}
			if (tradePay.getIsPaid().equals(TradeEnums.YesNoEnum.YES.getCode())) {
				throw new RuntimeException("该支付单已支付");
			}
			tradePay.setIsPaid(TradeEnums.YesNoEnum.YES.getCode());
			int i = tradePayMapper.updateByPrimaryKeySelective(tradePay);
			/*
			 * 当前rocketMQ不支持分布式事务，不能保证修改数据库和MQ消息同时成功或失败。 把MQ事务变成本地事务 --
			 * trade_mq_producer_temp
			 */
			if (i == 1) {
				final PaidMQ paidMQ = new PaidMQ();
				paidMQ.setPayAmount(tradePay.getPayAmount());
				paidMQ.setOrderId(tradePay.getOrderId());
				paidMQ.setPayId(tradePay.getPayId());
				final TradeMqProducerTemp mqProducerTemp = new TradeMqProducerTemp();
				mqProducerTemp.setId(IDGenerator.generatorUUID());
				mqProducerTemp.setGroupName("payProducerGroup");
				mqProducerTemp.setMsgKeys(tradePay.getPayId());
				mqProducerTemp.setMsgTag(MQEnums.TopicEnum.PAY_PAID.getTag());
				mqProducerTemp.setMsgBody(JSON.toJSONString(paidMQ));
				mqProducerTemp.setCreateTime(new Date());
				tradeMqProducerTempMapper.insert(mqProducerTemp);
				// 异步线程发送mq，发送成功清空发送表
				executorService.submit(new Runnable() {
					public void run() {
						try {
							SendResult sendResult = aceMQProducer.sendMessage(MQEnums.TopicEnum.PAY_PAID,
									paidMQ.getPayId(), JSON.toJSONString(paidMQ));
							System.out.println(sendResult);
							if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
								tradeMqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
								System.out.println("删除消息表成功");
							}
						} catch (AceMQException e) {
							e.printStackTrace();
						}
					}
				});
			} else {
				throw new RuntimeException("该支付单已支付");
			}
		}
		return callbackPaymentRes;
	}

}
